package com.gitee.sparta.core.cluster.netty;

import com.gitee.sparta.common.helpers.JsonHelper;
import com.gitee.sparta.core.bean.cluster.ClusterCmd;
import com.gitee.sparta.core.bean.cluster.ClusterProtocol;
import com.gitee.sparta.core.cluster.AbstractClusterFollower;
import com.gitee.sparta.core.cluster.AbstractClusterLeader;
import com.gitee.sparta.core.cluster.ClusterCache;
import com.gitee.sparta.core.cluster.balance.LoadBalance;
import com.gitee.sparta.core.cluster.netty.codec.ClusterProtocolCodec;
import com.gitee.sparta.core.cluster.netty.codec.ProtocolFrameDecoder;
import com.gitee.sparta.core.cluster.netty.handler.LeaderProtocolHandler;
import com.gitee.sparta.core.cluster.netty.handler.NettyHeartbeatHandler;
import com.gitee.sparta.core.process.feature.SpartaWorkerFeature;
import com.gitee.sparta.core.process.workers.SpartaCarryWorker;
import com.gitee.sparta.core.property.SpartaCoreProperties;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RedissonClient;

import java.net.InetSocketAddress;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;


/**
 * <p>
 *      netty server
 * </p>
 *
 * @author zyred
 * @since v 0.0.2
 */
@Slf4j
public class NettyClusterLeader extends AbstractClusterLeader {

    private final AtomicBoolean atomicBoolean = new AtomicBoolean(false);

    private Channel leaderChannel;
    private AbstractClusterFollower follower;

    public NettyClusterLeader(SpartaCoreProperties properties,
                              RedissonClient client,
                              LoadBalance loadBalance) {
        super(properties, client, loadBalance);
    }

    @Override
    @SuppressWarnings("all")
    public void buildLeader() {
        new Thread(this::buildNettyServer).start();
        while(!atomicBoolean.get()) {
            try {
                TimeUnit.MILLISECONDS.sleep(100);
            } catch (InterruptedException ex) {
                log.error("[Nio leader enable error] exception : {}", ex.getMessage(), ex);
            }
        };
    }


    @SuppressWarnings("all")
    private void buildNettyServer() {
        EventLoopGroup boos = new NioEventLoopGroup(1);
        EventLoopGroup worker = new NioEventLoopGroup(2);
        // leader 只负责读，不需要向 follower 写心跳数据
        final int rt = 60, wt = 0, at = 120;
        ServerBootstrap sb = new ServerBootstrap()
                .group(boos, worker)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline()
                                .addLast(new ProtocolFrameDecoder())
                                .addLast(new ClusterProtocolCodec())
                                .addLast(new LeaderProtocolHandler(properties, client, loadBalance))
                                // readerIdleTimeSeconds：15s 没有收到读动作，表示该通道出现网络波动，移除这个负载
                                .addLast(new IdleStateHandler(rt, wt, at))
                                .addLast(new NettyHeartbeatHandler());
                    }
                });

        try {
            // 启动
            ChannelFuture sync = sb.bind(0).sync();
            this.leaderChannel = sync.channel();
            // 从sync.channel().localAddress() 获取端口
            InetSocketAddress addr = (InetSocketAddress) sync.channel().localAddress();
            int serverPort = addr.getPort();
            // 保存启动所在的端口号
            super.properties.getCluster().setClusterTransportPort(addr.getPort());
            log.info("[Sprata] Netty Leader build success, ctp : {}", serverPort);
            // 启动完毕
            this.atomicBoolean.set(true);

            sync.channel().closeFuture().sync();

            log.info("[Sparta] Leader server close ...");
        } catch (InterruptedException ex) {
            log.error("[Sparta] Leader enable error, exception : {}", ex.getMessage(), ex);
        }
    }

    @Override
    public void doWriteFollower(ClusterProtocol protocol) {
        Channel channel = protocol.getChannel();
        if (Objects.nonNull(channel)) {
            channel.writeAndFlush(protocol);
        }
        else {
            log.info("[Sparta] Write follower, channel is null, protocol: {}",
                    JsonHelper.getJsonString(protocol));
        }
    }


    @Override
    public void demotionFollower(String ip, int port) throws Exception {
        /*
         * 1. 关闭搬运线程
         * 2. 关闭 leader 线程
         * 3. 广播给 此leader 下的 follower，让他们换 leader
         * 4. 关闭 server
         * 5. 连接新的 leader
         */
        log.info("[Sparta] Starting demotion follower ...");


        // 关闭搬运线程
        SpartaCarryWorker.enable = false;
        SpartaWorkerFeature.interruptWorker();

        // 关闭 leader 线程
        ClusterProtocol pro = ClusterProtocol.defaultValue(ClusterCmd.STOP_LEADER_SENDER);
        AbstractClusterLeader.sendQueue(pro);

        // 广播给 follower, 切换 leader
        Set<Channel> channels = ClusterCache.getNettyChannels();
        for (Channel channel : channels) {
            pro = ClusterProtocol.defaultValue(ClusterCmd.SWITCHOVER_LEADER);
            channel.writeAndFlush(pro);
        }

        // 关闭 leader
        this.leaderChannel.close();

        // 构建 follower
        this.follower.buildFollower(ip, port);

        log.info("[Sparta] Demotion follower success ...");
    }


    public void setFollower(AbstractClusterFollower follower) {
        this.follower = follower;
    }
}
